package com.b3434.Factory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.b3434.Bean.Parameter;
import com.b3434.Processor.*;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.log4j.Logger;


import java.util.ArrayList;
import java.util.List;

/**
 * Created by Lee on 2017/3/21 0021.
 */
public class ProcessorFactory {
    static Logger log = Logger.getLogger(ProcessorFactory.class);



    private static String schemaStr;

    public static String getSchemaStr() {
        return schemaStr;
    }

    /*
    * Create a new offline processor
    * @param parameterString 参数json字符串
    *
    * */
    public static Processor createOfflineProcessor(String parameterJsonStr){
        log.info("parameter=" + parameterJsonStr);
        Parameter parameter = JSONObject.parseObject(parameterJsonStr,Parameter.class);
        schemaStr = createOfflineSchema(parameter);
        log.info("OfflineSchemaStr="+schemaStr);
        return createOfflineProcessor(parameter, schemaStr);
    }

    /*
    * Create a new offline processor
    * @param parameter 参数对象
    * @param schemaJson shema Json string
    * */
    static Processor createOfflineProcessor(Parameter parameter, String schemaStr) {
        Schema.Parser parser=new Schema.Parser();
        Schema schema=parser.parse(schemaStr);
        Processor processor = new ConcreteProcessor(new GenericData.Record(schema),parameter);
        if (parameter.getIsAllString()){
            processor = new OriginProcessor(processor);
        }
        if (parameter.getIsRegexed()) {
            processor = new RegexProcessor(processor);
        }
        if (parameter.getIsSplited()) {
            processor = new SplitProcessor(processor);
        }
        if(parameter.getIsEncrypted()){
            processor = new EncryptProcessor(processor);
        }
        return processor;

    }

    /*
    * 新建一个Json数据处理器，需要的参数：
    *
    * @param parameter JsonString
    * @param JsonDemo 一个完整的Json的demo，深度为1
    *
    * @return 返回一个Processor
    * */
    public static Processor createJsonProcessor(String parameterJsonStr, String JsonDemo) {
        JSONObject jsonObjectDemo = JSONObject.parseObject(JsonDemo);
        Parameter parameter = JSONObject.parseObject(parameterJsonStr,Parameter.class);
        schemaStr = createJsonSchema(parameter, jsonObjectDemo);
        log.info("JsonSchemaStr="+schemaStr);
        return createJsonProcessor(parameter, schemaStr);
    }
    /*
    * 新建一个Json数据处理器，需要的参数：
    *
    * @param parameter parameter类
    * @param schema String schema的json String
    *
    * @return 返回一个Processor
    * */
    static Processor createJsonProcessor(Parameter parameter, String schemaStr) {
        Schema.Parser parser = new Schema.Parser();
        Schema schema = parser.parse(schemaStr);
        Processor processor = new ConcreteProcessor(new GenericData.Record(schema), parameter);
        processor = new JsonProcessor(processor);
        return processor;
    }

    /*
* 新建一个关系数据处理器，需要的参数：
*
* @param parameter JsonString
* @param JsonDemo 一个完整的Json的demo，深度为1
*
* @return 返回一个Processor
* */
    public static Processor createRelationshipProcessor(String parameterJsonStr, String JsonDemo) {
        JSONObject jsonObjectDemo = JSONObject.parseObject(JsonDemo);
        Parameter parameter = JSONObject.parseObject(parameterJsonStr,Parameter.class);
        schemaStr = createJsonSchema(parameter, jsonObjectDemo);
        log.info("JsonSchemaStr="+schemaStr);
        return createJsonProcessor(parameter, schemaStr);
    }
    /*
    * 新建一个关系数据处理器，需要的参数：
    *
    * @param parameter parameter类
    * @param schema String schema的json String
    *
    * @return 返回一个Processor
    * */
    static Processor createRelationshipProcessor(Parameter parameter, String schemaStr) {
        Schema.Parser parser = new Schema.Parser();
        Schema schema = parser.parse(schemaStr);
        Processor processor = new ConcreteProcessor(new GenericData.Record(schema), parameter);
        if(parameter.getWhiteList()==null)
        processor = new JsonProcessor(processor);
        else processor = new WhiteListProcessor(processor);
        return processor;
    }

    /*
    * Create a new field for fields of schema
    * @param name filedName
    * @param type type of data
    *
    * @return field
    *
    * */
    static JSONObject createNewFiledJsonObject(String name,String type){
        JSONObject jsonObject=new JSONObject();
        jsonObject.put("name", name);
        jsonObject.put("type","string");
        return jsonObject;
    }

    /*
* Create a Json data access schma for Avro
* @param Parameter 参数
* @param JSONObject jsondemo
* @return schema String
*
* */
    public static String createRelationshipSchema(Parameter parameter, JSONObject jsonObject) {
        return createJsonSchema(parameter,jsonObject);
    }
    /*
    * Create a Json data access schma for Avro
    * @param Parameter 参数
    * @param JSONObject jsondemo
    * @return schema String
    *
    * */
    public static String createJsonSchema(Parameter parameter, JSONObject jsonObject) {
        com.b3434.Bean.Schema schema = new com.b3434.Bean.Schema();
        schema.setNamespace("com.taotie.json").setType("record").setName(parameter.getTopic());
        List<JSONObject> fields = new ArrayList<JSONObject>();
        for (String i : jsonObject.keySet()) {
            fields.add(createNewFiledJsonObject(i, "string"));
        }
        schema.setFields(fields);
//        log.info(JSON.toJSONString(schema));
        return JSON.toJSONString(schema);
    }
    /*
    * Create a Offline data access schma for Avro
    * @param Parameter 参数
    *
    * @return schema String
    *
    * */
    public static String createOfflineSchema( Parameter parameter) {
        com.b3434.Bean.Schema schema = new com.b3434.Bean.Schema();
        schema.setNamespace("com.taotie.offline").setType("record").setName(parameter.getTopic());


        List<JSONObject> fields = new ArrayList<JSONObject>();

        if (parameter.getIsAllString()){
            fields.add(createNewFiledJsonObject("originstr","string"));
        }
        if (parameter.getIsRegexed()){
            for (String str:parameter.getFiledNameRegexed()){
                fields.add(createNewFiledJsonObject(str,"string"));
            }
        }
        if (parameter.getIsSplited()){
            for (String str:parameter.getFiledNameSplitedList()){
                fields.add(createNewFiledJsonObject(str,"string"));
            }
        }
        schema.setFields(fields);
//        log.info(JSON.toJSONString("schema="+schema));

        return JSON.toJSONString(schema);
    }


}
